package com.cn.broker;

import com.cn.common.ClientInfo;
import com.cn.common.consumer.ClientChannelInfo;
import com.cn.common.consumer.ConsumerGroupInfo;
import io.netty.channel.Channel;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

public class ConsumerManager {

    private ConcurrentMap<String /*group*/, ConsumerGroupInfo> consumerTables = new ConcurrentHashMap<>();

    public void register(ClientInfo clientInfo, Channel channel) {
        if (clientInfo == null) {
            return;
        }
        ConsumerGroupInfo consumerGroupInfo = this.consumerTables.get(clientInfo.getGroup());
        if (null == consumerGroupInfo) {
            consumerGroupInfo = new ConsumerGroupInfo().setConsumerGroup(clientInfo.getGroup());
            this.consumerTables.put(clientInfo.getGroup(), consumerGroupInfo);
        }
        if (consumerGroupInfo != null) {
            consumerGroupInfo.updateChannel(channel, clientInfo);
        }
    }

    public List<String> filterGroup(String group) {
        List<String> filterList = new ArrayList<>();
        ConsumerGroupInfo consumerGroupInfo = this.consumerTables.get(group);
        if (consumerGroupInfo != null) {
            for (ClientChannelInfo clientChannelInfo : consumerGroupInfo.getClientChannelTabls().values()) {
                filterList.add(clientChannelInfo.getClientId());
            }
        }
        return filterList;
    }

    public void unregister(Channel channel) {
        for (ConsumerGroupInfo consumerGroupInfo : consumerTables.values()) {
            ClientChannelInfo pre = consumerGroupInfo.getClientChannelTabls().remove(channel);
            if (pre != null) {
                //cid改变了
                System.out.println("cid改变了");
            }
        }
    }

}
